#include "dispatch_server.h"

int dispatch_listener_handler::read(socket_event *se)
{
	address addr;
	socklen_t length = addr.length();
	while (1) {
		int sock = accept(se->get_handle(), &addr, &length);
		if (sock < 0) {
			if ((errno == EAGAIN || errno == EWOULDBLOCK)) {
				return 3103;
			} else {
				LOG_ERROR_VA("[%s] accept error", _svr->whoami().c_str());
				return -1;
			}
		}
		LOG_DEBUG_VA("[%s] accept <%s,%d> @socket:%d", _svr->whoami().c_str(), addr.ip().c_str(), addr.port());
		socket_event *client_sock_event = new socket_event(socket_event::read);
		client_sock_event->set_handle(sock);
		client_sock_event->set_event_handler(new dispatch_client_handler(_svr));
		if (_svr->add_client_socket_event(client_sock_event) != 0) {
			delete client_sock_event;
			LOG_ERROR_VA("[%s] add client error, maybe event-pool full", _svr->whoami().c_str());
		}
	}
	return 3103;
}

int dispatch_client_handler::read(socket_event *se)
{
	error_response_message emsg;
	LOG_INFO_VA("[%s] new request message coming in...", _svr->whoami().c_str());
	head msg_head;
	int sock = se->get_handle();
	int ret = tcp_read_ms(sock, &msg_head, sizeof(msg_head), tcp_read_timeout);
	if (ret <= 0) {
		LOG_ERROR_VA("[%s] read head error or client closed", _svr->whoami().c_str());
		return ret;
	}
	if (ret != sizeof(msg_head)) {
		LOG_ERROR_VA("[%s] bad request, refused to service", _svr->whoami().c_str()); 
		return -1;
	}
	msg_head.ntoh();
	LOG_INFO_VA("[%s] likely msg head, cmd:%d, vr:%d, src:%d, st:%d, len:%d, sn:%llu-%llu", _svr->whoami().c_str(), 
		msg_head.command, msg_head.version, msg_head.src, msg_head.status, msg_head.length, 
		msg_head.sn_high, msg_head.sn_low);

	buffer &buf = _svr->get_buffer();
	ret = tcp_read_ms(sock, buf, msg_head.length, tcp_read_timeout);

	if (ret != msg_head.length) {
		LOG_ERROR_VA("[%s] message body receive failed, recv:%d bytes, expect:%d, maybe bad format",
		_svr->whoami().c_str(), ret, msg_head.length);
		emsg.set_error_str("bad msg body length");
		send_msg(se->get_handle(), emsg);
		return -1;
	}
	pure_message *msg = new pure_message;
	msg->set_head(msg_head);
	msg->set_buffer(buf);
	msg->set_socket_event(se);
	task *tsk = new msg_handle_task(msg);
	LOG_DEBUG_VA("[%s] create msg task for msg %d", _svr->whoami().c_str(), msg->get_id());
	tsk->start(_svr->get_thread_pool());
	return 3103;
}

int pure_message::response(worker *wkr)
{
	dispatch_worker *w = static_cast<dispatch_worker *>(wkr);
	int cmd = get_head().command; 
	pure_message resp;

	server_proxy *front = w->choose_one_front_node();
	if(front){
		if (front->send_msg(*this) <= 0) {
			LOG_ERROR_VA("front request send msg error");
			response_error("front request send msg error");
			front->reset();
			return -1;
		}
	
		if (front->recv_msg(resp) <= 0) {
			LOG_ERROR_VA("front request recv msg error");
			response_error("front request recv msg error");
			front->reset();
			return -1;
		}

	}else{
		LOG_ERROR_VA("no front choosed");
		response_error("no front choosed");
		return -1;
	}
/*
	if(resp.get_status() == 0 && (get_head().sn_high != resp.get_head().sn_high || get_head().sn_low != resp.get_head().sn_low)){
		LOG_ERROR_VA("[dispatch] sn error");
	}
*/
	send_msg(this->get_handle(), resp);
	return 0;
}

int dispatch_server::setup()
{
	this->set_whoami("dispatch");
	dispatch_worker *w = new dispatch_worker;
	vector<address>::iterator it = _front_servers.begin();
	vector<int>::iterator wit = _front_servers_weight.begin();
	for (; it != _front_servers.end(); it++, wit++)
		w->add_front_node(*it, *wit);

	this->set_msg_worker(w);
	return 0;
}

